import json
import redis
from views.send_view import send_index, send_wave


class Handels:
    def __init__(self, kafka_producer, minio_producer):
        self.kafka_producer = kafka_producer
        self.minio_producer = minio_producer
        self.redis_client = redis.StrictRedis(
            host='127.0.0.1',
            port=19007,
            password=123456,
            decode_responses=True,  # 关键参数
            charset='utf-8',  # 默认编码
            db=1
        )

    def case_to_function(self, topic_type):
        func_name = f'handel_{topic_type}'
        if hasattr(self, func_name):
            func = getattr(self, func_name)
            return func

    async def handel_IndexData(self, payload, topic_sn):
        res = eval(payload.decode('utf-8'))
        await send_index(res, topic_sn, self.kafka_producer)

    async def handel_WaveData(self, payload, topic_sn):
        res = eval(payload.decode('utf-8'))
        await send_wave(res, topic_sn, self.redis_client, self.kafka_producer, self.minio_producer)


class RespDto:
    topic = ""
    payload = []
    is_replay = True

    def __init__(self, topic, payload) -> None:
        self.topic = topic
        self.payload = payload


async def handel(topic, payload, kafka_producer, minio_producer):
    """
    示例化信息处理函数
    :param topic: 主题全称
    :param payload: 消息内容
    :param topic_type: 主题类型
    :param topic_sn: 主题中的采集器编码
    :return: 返回执行状态
    """
    status = "1"
    # topic_type: 主题类型, topic_sn: 主题中的采集器编码
    topic_type, topic_sn = topic[3], topic[2]
    res = RespDto(topic, '')
    try:
        func = Handels(kafka_producer, minio_producer).case_to_function(topic_type)
        if func is not None:
            res = await func(payload, topic_sn)
        else:
            status = "0"
            print(f'主题：{topic} 失败，不支持的主题类型: {topic_type}')
    except Exception as e:
        status = "0"
        print(f'主题：{topic} 失败，{e}')
    res.topic += '/' + status
    print(topic + " " + json.dumps(res.__dict__))
